package cdc;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/26
 */
public class Mysql2Mysql {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);

        bsTableEnv.executeSql("create table hudi_test (id int,  name String, age int, PRIMARY KEY (`id`) NOT ENFORCED) with ('connector' = 'mysql-cdc'," +
                "'hostname' = '10.10.100.161', 'port' = '3306', 'database-name' = 'xyc_test_cdc','table-name' = 'hudi_test', 'username'='root'," +
                "'password'='supconit161','scan.incremental.snapshot.enabled' = 'false')");


        bsTableEnv.executeSql("create table hudi_test2 (id int,  addr String, PRIMARY KEY (`id`) NOT ENFORCED) with ('connector' = 'mysql-cdc'," +
                "'hostname' = '10.10.100.161', 'port' = '3306', 'database-name' = 'xyc_test_cdc','table-name' = 'hudi_test_2', 'username'='root'," +
                "'password'='supconit161', 'scan.incremental.snapshot.enabled' = 'false')");

        bsTableEnv.executeSql("create table hudi_test3 (id int, name String, age int,addr String, PRIMARY KEY (`id`) NOT ENFORCED) with ('connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://10.10.100.161:3306/xyc_test_cdc', 'table-name' = 'hudi_test3', 'username'='root'," +
                "'password'='supconit161')");

        bsTableEnv.executeSql("insert into hudi_test3 select ht.id, ht.name, ht.age, ht2.addr from hudi_test ht join hudi_test2 ht2 on ht.id = ht2.id");

    }
}
